package com.tornainbow;

import java.io.ByteArrayInputStream;
import java.nio.charset.Charset;
import java.util.Arrays;
import java.util.Scanner;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author mrcrysl
 */
public class DelimiterParser extends Processor {
    // Define a logger for this class
    private static Logger logger = LoggerFactory.getLogger(DelimiterParser.class);
    
    // Define the required properties
    public static final String PROP_CHARSET = "charset";
    public static final String PROP_SKIP_NUM_LINES = "skip.num.lines";
    public static final String PROP_COLUMN_HEADERS = "column.headers";
    public static final String PROP_DELIMITER = "delimiter";
    
    // Define default values for the properties
    private static final String DEFAULT_CHARSET = Charset.defaultCharset().name();
    private static final int DEFAULT_SKIP_NUM_LINES = 0;
    private static final String DEFAULT_COLUMN_HEADERS = "";
    private static final String DEFAULT_DELIMITER = ",";
    
    // Define the set of metacharacters that need to be escaped for the reqex expression
    private static final ConcurrentMap<String,String> METACHARACTERS = new ConcurrentHashMap<>();
    static {
        METACHARACTERS.put("[", "\\[");
        METACHARACTERS.put("\\", "\\\\");
        METACHARACTERS.put("^", "\\^");
        METACHARACTERS.put("$", "\\$");
        METACHARACTERS.put(".", "\\.");
        METACHARACTERS.put("|", "\\|");
        METACHARACTERS.put("?", "\\?");
        METACHARACTERS.put("*", "\\*");
        METACHARACTERS.put("+", "\\+");
        METACHARACTERS.put("(", "\\(");
        METACHARACTERS.put(")", "\\)");
    }
    
    // Define the characater that may be enclosing none, some or all the fields
    private static final char ENCLOSING_CHARACTER = '"';
    
    // Define the system-dependent line separator
    private static final String LINE_SEPARATOR = System.getProperty("line.separator");
    
    /**
     * Set the properties for this class
     */
    public DelimiterParser() {
        super();
        properties.putIfAbsent(PROP_CHARSET, DEFAULT_CHARSET);
        properties.putIfAbsent(PROP_DELIMITER, DEFAULT_DELIMITER);
        properties.putIfAbsent(PROP_COLUMN_HEADERS, DEFAULT_COLUMN_HEADERS);
        properties.putIfAbsent(PROP_SKIP_NUM_LINES, String.valueOf(DEFAULT_SKIP_NUM_LINES));
    }
    
    @Override
    public String getProcessorName() {
        return (this.getClass().getName() + " : " + this.getName());
    }
    
    @Override
    public void run() {
        // Get the character set from the properties map
        String charset = this.getProperty(PROP_CHARSET);
        logger.debug("Setting charset to {}", charset);

        // Get the number of lines to skip from the properties map
        int skipNumLines = DEFAULT_SKIP_NUM_LINES;
        try {
            skipNumLines = Integer.parseInt(this.getProperty(PROP_SKIP_NUM_LINES));
        } catch (NumberFormatException e) {
            logger.warn("Failed to parse the value {} as an Integer...using the default value {}", this.getProperty(PROP_SKIP_NUM_LINES), DEFAULT_SKIP_NUM_LINES);
        }
        logger.debug("Setting skipNumLines to {}", skipNumLines);

        // Get the column headers from the properties map
        String columnHeadersVal = this.getProperty(PROP_COLUMN_HEADERS);
        logger.debug("Setting columnHeadersVal to {}", columnHeadersVal);

        // Get the delimiter from the properties map
        String delimiter = this.getProperty(PROP_DELIMITER);
        // Check to see if the delimiter is a known METACHARACTER and if so, replace
        // it with its counter-part
        if (METACHARACTERS.get(delimiter) != null) {
            delimiter = METACHARACTERS.get(delimiter);
        }
        logger.debug("Setting delimiter to {}", delimiter);

        // Loop indefinitely waiting for new messages
        for (;;) {
            // Wait to receive a new message
            Record record = Consumer.receive(this.getProperty(PROP_JMS_URL), this.getProperty(PROP_JMS_SOURCE_QUEUE_NAME));
            logger.info("Received message from queue {}", this.getProperty(PROP_JMS_SOURCE_QUEUE_NAME));

            // Check to make sure the Record object is not null
            if (record == null) {
                logger.error("Received null message from the queue...skipping");
                continue;
            }

            // Convert the payload of the Record object into a byte array input stream
            ByteArrayInputStream bais = new ByteArrayInputStream(record.getPayload());
            logger.debug("Converted payload into ByteArrayInputStream object");

            // Instantiate a new Scanner object to loop over the lines of date
            Scanner scanner = new Scanner(bais, charset);
            logger.debug("Instantiated Scanner object using charset {}", charset);

            // Assume each line ends with the system-dependent line separator
            scanner.useDelimiter(LINE_SEPARATOR);
            logger.debug("Set Scanner object to use delimiter {}", LINE_SEPARATOR);

            // Skip the specified number of lines
            logger.debug("Skipping {} lines in the file", skipNumLines);
            for (int i=0; i < skipNumLines; i++) {
                if (scanner.hasNext()) {
                    scanner.next();
                }
            }
            
            // Get the column headers if they weren't specified by a property
            String[] columnHeaders = null;
            if (columnHeadersVal != null && !columnHeadersVal.trim().isEmpty()) {
                columnHeaders = columnHeadersVal.split(",");
                logger.debug("Setting columnHeaders to {}", columnHeadersVal);
            } else {
                logger.debug("Did not receive column headers...will assume they are provided by the data");
            }

            // If no column headers have been specified then assume the first line
            // in the file is providing them
            if (columnHeaders == null || columnHeaders.length == 0) {
                logger.debug("No column headers were provided...using the first line in the file");
                // Get the first line from the file and process it
                String line = null;
                if (scanner.hasNext()) {
                    line = scanner.next();
                    columnHeaders = this.processLine(line, delimiter);
                    logger.debug("Retrieved {} column headers from the data", columnHeaders.length);
                }
            }

            // Check to make sure at least some column headers were processed
            if (columnHeaders == null || columnHeaders.length == 0) {
                logger.error("Failed to process any column headers...unable to continue");
                return;
            }

            // Loop over each line from the file and process it
            while (scanner.hasNext()) {
                // Create a new Record and keep the payload from the original record
                Record newRecord = new Record();
                newRecord.setPayload(record.getPayload());
                logger.debug("Created new Record object from the original Record object");

                // Get the array of values from the line
                String[] values = this.processLine(scanner.next(), delimiter);
                logger.debug("Received {} values from the line", values.length);

                // Check to make sure the number of values does not exceed the
                // number of column headers
                if (values.length > columnHeaders.length) {
                    logger.error("The number of values {} is greater than the number of column headers {}...skipping line", values.length, columnHeaders.length);
                    continue;
                }

                // Write the key/value pairs to the output stream
                for (int i=0; i < columnHeaders.length; i++) {
                    String key = columnHeaders[i];
                    String value = values[i];
                    newRecord.addField(new Field(key, value));
                    logger.debug("Added key/value pair: {} = {}", key, value);
                }

                // Send the message to the queue
                Producer.sendMessage(this.getProperty(PROP_JMS_URL), this.getProperty(PROP_JMS_DESTINATION_QUEUE_NAME), newRecord);
                logger.info("Sent message to the queue {}", this.getProperty(PROP_JMS_DESTINATION_QUEUE_NAME));
            }
        }
    }
    
    /**
     * Extracts a set of values from the given line using the specified delimiter
     * @param line The line to process
     * @param delimiter The delimiter to use that should separate fields
     * @return An array of values extracted from the given line
     */
    private String[] processLine(String line, String delimiter) {
        logger.debug("Processing line: {}", line);
        
        // Keep track of the type of value being processed
        boolean processingEnclosedValue = false;
        boolean processingNonEnclosedValue = false;
        
        // Keep track of all the values processed by this line
        String[] valuesProcessed = new String[0];
        
        // Define a variable to hold a value as it is processed
        StringBuilder value = new StringBuilder();
        
        // Keep track of the number of times we see the delimiter as it might
        // indicate an empty value
        int numTimesSeen = 0;
        
        // Loop over each character in the line
        for (int i=0; i < line.length(); i++) {
            // Get the next character
            char c = line.charAt(i);
            
            // If we are not currently processing a field and the character equals
            // the ENCLOSING_CHARACTER then change the enclosed processing flag to
            // true to indicate that a field is about to be processed
            if (!processingEnclosedValue && c == ENCLOSING_CHARACTER) {
                processingEnclosedValue = true;
                numTimesSeen = 0;
                continue;
            }
            
            // If we are currently processing an enclosed field and the character 
            // does not equal the ENCLOSING_CHARACTER then append it to the value
            if (processingEnclosedValue && c != ENCLOSING_CHARACTER) {
                value.append(c);
                continue;
            }
            
            // If we are currently processing an enclosed field and the character
            // equals the ENCLOSING_CHARACTER then assume we have reached the end
            // of a field so change the enclosed processing flag to false, add the
            // value of the field to the value array and reset the value variable
            if (processingEnclosedValue && c == ENCLOSING_CHARACTER) {
                logger.debug("Consumed enclosed field value {}", value);
                processingEnclosedValue = false;
                valuesProcessed = Arrays.copyOf(valuesProcessed, valuesProcessed.length+1);
                valuesProcessed[valuesProcessed.length-1] = value.toString().trim();
                value = new StringBuilder();
                continue;
            }
            
            // If we are not currently processing an enclosed field and we are not
            // currently processing a non-enclosed field and the character equals 
            // the delimiter then we need to perform an additional check
            if ((!processingEnclosedValue && !processingNonEnclosedValue) && Character.toString(c).matches(delimiter)) {
                // If this is the second time we've seen the delimiter then it we can
                // assume we just consumed an empty value so update the valuesProcessed
                // array accordingly and subtract one from the numTimesSeen variable
                if (++numTimesSeen == 2) {
                    logger.debug("Consumed empty value");
                    valuesProcessed = Arrays.copyOf(valuesProcessed, valuesProcessed.length+1);
                    valuesProcessed[valuesProcessed.length-1] = value.toString().trim();
                    value = new StringBuilder();
                    numTimesSeen--;
                }
                // Perform one last check in case the last character in the line is also
                // the delimiter
                if ((i+1) == line.length()) {
                    c = line.charAt(line.length()-1);
                    if (Character.toString(c).matches(delimiter)) {
                        logger.debug("Consumed empty value");
                        valuesProcessed = Arrays.copyOf(valuesProcessed, valuesProcessed.length+1);
                        valuesProcessed[valuesProcessed.length-1] = value.toString();
                        value = new StringBuilder();
                        numTimesSeen = 0;
                    }
                }
                continue;
            }
            
            // If we are not currently processing a non-enclosed field and the character
            // does not equal the delimiter then change the non-enclosed processing flag
            // to true and append the character to the value
            if (!processingNonEnclosedValue && !Character.toString(c).matches(delimiter)) {
                processingNonEnclosedValue = true;
                value.append(c);
                continue;
            }
            
            // If we are currently processing a non-enclosed field and the character does
            // not equals the delimiter then append the character to the value
            if (processingNonEnclosedValue && !Character.toString(c).matches(delimiter)) {
                value.append(c);
                // If this was the last character in the line then we need to consume the value
                if ((i+1) == line.length()) {
                    logger.debug("Consumed non-enclosed field value {}", value);
                    processingNonEnclosedValue = false;
                    valuesProcessed = Arrays.copyOf(valuesProcessed, valuesProcessed.length+1);
                    valuesProcessed[valuesProcessed.length-1] = value.toString().trim();
                    value = new StringBuilder();
                    numTimesSeen = 0;
                }
                continue;
            }
            
            // If we are currently processing a non-enclosed field and the character 
            // equals the delimiter then assume we have reached the end of a field so 
            // change the non-enclosed processing flag to false, add the value of the 
            // field to the value array, reset the value variable and set the number
            // of times we have seen the delimiter to one
            if (processingNonEnclosedValue && Character.toString(c).matches(delimiter)) {
                logger.debug("Consumed non-enclosed field value {}", value);
                processingNonEnclosedValue = false;
                valuesProcessed = Arrays.copyOf(valuesProcessed, valuesProcessed.length+1);
                valuesProcessed[valuesProcessed.length-1] = value.toString().trim();
                value = new StringBuilder();
                numTimesSeen = 1;
                continue;
            }
        }
        
        // Return the array of values that were processed
        return valuesProcessed;
    }
}
